package com.mfs.proxy.core.server.reactor;

import com.mfs.common.utills.ChannelUtil;
import com.mfs.proxy.core.exceptions.ProxyException;
import com.mfs.proxy.core.client.ProxyClient;
import com.mfs.proxy.core.server.config.ServerConfig;
import com.mfs.proxy.core.server.registry.ProxyClientRegistrySingletonFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.Set;

/**
 * 用户通道Reactor
 * */
public class SubProxyReactor implements Runnable {
    private static Selector subProxySelector;
    private final static Logger logger = LoggerFactory.getLogger(SubProxyReactor.class);
    public SubProxyReactor() {
        try {
            subProxySelector = Selector.open();
        } catch (IOException e) {
            throw new ProxyException(e);
        }
    }

    public static SelectionKey register(SelectableChannel selectableChannel) throws ClosedChannelException {
        return selectableChannel.register(subProxySelector.wakeup(), SelectionKey.OP_READ);
    }

    @Override
    public void run() {
        while (!Thread.currentThread().isInterrupted()) {
            try {
                int select = subProxySelector.select(100);
                if (select < 1) {
                    Thread.yield();
                    continue;
                }
                Set<SelectionKey> selectionKeys = subProxySelector.selectedKeys();
                Iterator<SelectionKey> iterator = selectionKeys.iterator();
                while (iterator.hasNext()) {
                    SelectionKey selectionKey = iterator.next();
                    processReadable(selectionKey);
                    iterator.remove();
                }
            } catch (IOException e) {
                logger.error("处理用户请求失败", e);
            }
        }
    }

    /**
     * 处理用户请求，其他忽略
     * */
    private void processReadable(SelectionKey selectionKey) {
        if (!(selectionKey.isValid() && selectionKey.isReadable())) {
            return;
        }
        // 获取代理客户端
        SocketChannel userChannel = (SocketChannel) selectionKey.channel();
        int proxyPort = userChannel.socket().getLocalPort();
        ProxyClient proxyClient = ProxyClientRegistrySingletonFactory.getProxyClientRegistry().checkout(ServerConfig.getInstance().getProxyIp(), proxyPort);
        selectionKey.cancel();
        // 处理
        if (proxyClient != null) {
            proxyClient.processRequest(userChannel);
        } else {
            // 说明代理客户端以被注销，只是由于ProxyClient还未被回收，所以资源未被释放
            ChannelUtil.closeSocketChannel((SocketChannel) selectionKey.channel());
        }
    }
}
